package com.example.aicloud.controller;

import com.example.aicloud.entity.Answer;
import com.example.aicloud.util.AppVariable;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;

@RestController
@RequestMapping("/kafka")
public class KafkaController {
    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Resource
    private ObjectMapper objectMapper;
    private static final String CANAL_TOPIC = "ai-cloud-canal-to-kafka";

    @KafkaListener(topics = CANAL_TOPIC)
    public void canalListen(String data, Acknowledgment ack) throws JsonProcessingException {
        HashMap<String,Object> map = objectMapper.readValue(data, HashMap.class);
        if (!map.isEmpty() && (map.get("database").toString().equals("aicloud") || map.get("database").toString().equals("aicloud2")) && map.get("table").toString().equals("answer")){
            // 更新Redis缓存
            ArrayList<LinkedHashMap<String,Object>> list = (ArrayList<LinkedHashMap<String,Object>>) map.get("data");
            String cacheKey = "";
            for (LinkedHashMap<String,Object> answer : list){
                cacheKey = AppVariable.getListCacheKey(Long.parseLong(answer.get("uid").toString()),Integer.parseInt(answer.get("model").toString()) , Integer.parseInt(answer.get("type").toString()));
                redisTemplate.opsForValue().set(cacheKey,null);
            }
        }
        ack.acknowledge(); //手动确认应答
    }
}
